/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.nhncorp.neptune.changelog.pipe;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class FileAppendThread extends Thread {
  private static final Log LOG = LogFactory.getLog(FileAppendThread.class);
  private static final long MAX_QUEUED_COUNT = 50;
  
  Object emptyCond = new Object();
  Object fullCond = new Object();
  AtomicInteger currentCount = new AtomicInteger(0);

  ConcurrentLinkedQueue<LogAppendAction> queue = new ConcurrentLinkedQueue<LogAppendAction>();
  boolean shutdown = false;
  
  public void register(ChangeLogFileChannel ch, ByteBuffer[] bufferList) throws IOException {
    /*
    int bufSize = 0;
    for(ByteBuffer buf : bufferList) {
      bufSize += buf.limit();
    }
    
    byte[] buffer = new byte[bufSize];
    
    
    while(currentCount.get() >= MAX_QUEUED_COUNT) {
      long before = System.currentTimeMillis();
      synchronized(fullCond) {
        try {
          LOG.info("wait");
          fullCond.wait();
        } catch(Exception e) {
        }
      }
      
      if (currentCount.get() < MAX_QUEUED_COUNT) {
        LOG.info("wake up after sleeping " + (System.currentTimeMillis() - before) + " ms");
      }
    }
    
    queue.add(new LogAppendAction(ch, bos.toByteArray()));

    if (currentCount.getAndIncrement() == 0) {
      synchronized(emptyCond) {
        emptyCond.notifyAll();
      }
    }
    */
  }

  public void run() {
    while(true) {
      while(currentCount.get() <= 0) {
        synchronized(emptyCond) {
          try {
            emptyCond.wait();
          } catch(Exception e) {
          }
        }
      }
      
      queue.poll().append();
      
      if (currentCount.decrementAndGet() < MAX_QUEUED_COUNT) {
        synchronized(fullCond) {
          fullCond.notifyAll();
        }
      }
    }
  }
}

class LogAppendAction {
  private static final Log LOG = LogFactory.getLog(LogAppendAction.class);

  ChangeLogFileChannel ch;
  byte[] byteArray;
  
  public LogAppendAction(ChangeLogFileChannel ch, byte[] byteArray) {
    this.ch = ch;
    this.byteArray = byteArray;
  }

  public long append() {
//    try {
//      ch.write(byteArray);
//      return byteArray.length;
//    } catch(IOException e) {
//      LOG.error("Writing log file is fail", e);
//    }
//    
    return -1;
  }
}